package cn.xshi.iot.common.mqtt;
import cn.xshi.common.util.StringUtil;
import cn.xshi.iot.common.vo.MqttDTO;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
 * @Desc
 * @Author 邓纯杰
 * @CreateTime 2012-12-12 12:12:12
 */
@Component
@Slf4j
public class MQTTSubsribe implements CommandLineRunner {

    private MqttConnectOptions mqttConnectOptions;

    private ScheduledExecutorService scheduled;

    private MqttClient client;

    MqttDTO mqttDTO = new MqttDTO();

    @Resource
    private MQTTSubsribeCallback pushCallback;

    public void startReconnect(){
        this.scheduled = Executors.newSingleThreadScheduledExecutor();
        // 定时任务——重新连接mqtt服务器
        this.scheduled.scheduleAtFixedRate(new Runnable() {
            public void run() {
                if (!MQTTSubsribe.this.client.isConnected()) {
                    try {
                        MQTTSubsribe.this.client.connect(MQTTSubsribe.this.mqttConnectOptions);
                        int[] Qos  = new int[mqttDTO.getSubTopic().length];
                        MQTTSubsribe.this.client.subscribe(mqttDTO.getSubTopic(), Qos);
                    } catch (MqttSecurityException var2) {
                        log.error("MqttSecurityException，异常信息："+var2.getLocalizedMessage());
                    } catch (MqttException var3) {
                        log.error("MqttSecurityException，异常信息："+var3.getLocalizedMessage());
                    }
                }
            }
        },1000L,5000L, TimeUnit.MILLISECONDS);
    }

    /**
     *
     * @return
     */
    public MqttConnectOptions getMqttConnectOptions(){
        this.mqttConnectOptions = new MqttConnectOptions();
        this.mqttConnectOptions.setCleanSession(true);
        this.mqttConnectOptions.setUserName(mqttDTO.getUserName());
        this.mqttConnectOptions.setPassword(mqttDTO.getPassword().toCharArray());
        this.mqttConnectOptions.setServerURIs(new String[]{mqttDTO.getHost()});
        this.mqttConnectOptions.setConnectionTimeout(10);
        this.mqttConnectOptions.setKeepAliveInterval(2000);
        return mqttConnectOptions;
    }

    /**
     * 连接mqtt服务器订阅信息方法
     */
    public void start() {
        try {
            if(StringUtil.isEmpty(mqttDTO.getHost()) ||
                    StringUtil.isEmpty(mqttDTO.getUserName())
            || StringUtil.isEmpty(mqttDTO.getPassword())){
                return;
            }
            String clientId= mqttDTO.getClientId();
            this.client = new MqttClient(mqttDTO.getHost(), clientId, new MemoryPersistence());
            this.getMqttConnectOptions();
            this.client.setCallback(this.pushCallback);
            this.client.connect(this.mqttConnectOptions);
            //订阅消息
            int[] Qos  = new int[mqttDTO.getSubTopic().length];
            this.client.subscribe(mqttDTO.getSubTopic(), Qos);
            boolean connected = this.client.isConnected();
            log.info("连接状态为:"+connected);
        } catch (MqttException e) {
            log.error("ReportMqtt客户端连接异常，异常信息："+e);
        }
    }

    /**
     *
     * @param args
     * @throws Exception
     */
    public void run(String... args) throws Exception {
        start();
    }

    /**
     * 取消订阅
     */
    public void cleanTopic(String topic) {
        if (null != client || client.isConnected()) {
            try {
                client.unsubscribe(topic);
            } catch (MqttException e) {
                e.printStackTrace();
                log.info("mqtt unsubscribe error,{}",e);
            }
        } else {
            log.info("mqttClient is error");
        }
    }
}
